package com.bgq.trxpay.listener;


import com.alibaba.fastjson2.JSONObject;
import com.bgq.trxpay.mapper.MessageRecordMapper;
import com.bgq.trxpay.model.MessageRecord;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class MessageListener {


    @Autowired
    private MessageRecordMapper messageRecordMapper;


    @KafkaListener(topics = {"fox"}, id = "bgq-1")
    public void onMessage(ConsumerRecord<?, ?> record, Acknowledgment acknowledgment) {
        // 消费的哪个topic、partition的消息,打印出消息内容
        String msg = record.topic() + "-" + record.partition() + "-" + record.value();
        System.out.println(msg);

        Object value = record.value();
        JSONObject jsonData = JSONObject.parseObject(value.toString());
        String msgId = jsonData.getString("msgId");

        // update message record table
        MessageRecord messageRecord = new MessageRecord();
        messageRecord.setId(Long.valueOf(msgId));
        messageRecord.setTrxStatus("1");
        messageRecordMapper.update(messageRecord);


        // 如果开启手动确认的话，那需要 在配置文件中 配置  listener  的设置,ack-mode
        acknowledgment.acknowledge();
    }

}
